package cn.serein.work;

import com.rabbitmq.client.*;

import java.io.IOException;

/**
 设置队列持久化
 */
public class WorkConsumer {
    public static void main(String[] args) throws Exception {
        //1.连接到RabbitMQ服务器
        ConnectionFactory cf = new ConnectionFactory();
        //利用连接工厂设置连接信息
        cf.setHost("192.168.8.100");
        cf.setPort(5672);//可选，5672是默认端口
        cf.setUsername("admin");
        cf.setPassword("Aq71056r");
        //2.创建连接对象（TCP）---所有的RabbitMQ服务器的客户端使用一个tcp连接对象
        Connection nc = cf.newConnection();
        //3.利用tcp连接对象创建channel（通道）
        Channel cc = nc.createChannel();
        //4.通过channel创建队列
        //第一个参数：队列名为simple.queue
        //第二个参数：队列是否持久化，将第二个参数设置为true，表示队列持久化
        //第三个参数：队列是否独占
        //第四个参数：队列是否自动删除
        //第五个参数：其他属性的设置
        cc.queueDeclare("work.queue", true, false, false, null);
        //处理数据的回调函数
        DeliverCallback d = new DeliverCallback() {
            @Override
            public void handle(String consumerTag, Delivery message) throws IOException {
                //Delivery可以获取要消费的数据
                System.out.println(consumerTag);
                byte[] a = message.getBody();
                String s = new String(a);
                //处理消息
                System.out.println(consumerTag + "收到：" + s);
                //模拟消息处理缓慢，如果输入的消息包含.那就睡眠10s钟
                if (s.contains(".")) {
                    try {
                        System.out.println(consumerTag + "开始处理消息...");
                        //睡眠10s钟
                        Thread.sleep(10000);
                        System.out.println(consumerTag + "处理消息完毕!");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println("-----------");
                //第一个参数：消息的标签，需要从消息中获取
                //是否确认多条消息处理完毕，false代表只确认一条消息
                cc.basicAck(message.getEnvelope().getDeliveryTag(), false);
            }
        };
        //取消接收时的回调函数
        CancelCallback c = new CancelCallback() {
            @Override
            public void handle(String consumerTag) throws IOException {

            }
        };
        //如果想实现消息的合理分发，还需要设置一个消费者同时只能处理一条消息
        cc.basicQos(1);
        //第一个参数：指定队列名称
        //第二个参数：先设置为true,设置为Autoack自动确认消息，只要消息发送给消费者
        //就会认为自己的消息处理完毕了，从RabbitMQ中删除，如果消费者还没来的及消费，就宕机了
        //就会造成消息丢失，所以我们不使用自动ack，将第二个值改为false
        //此处设置为false，还没完成ack的确认，消息消费完毕还需还设置ack的主动提交
        //手动ack的模式下：如果消费者还未发送消息回执的标签，这个消息会暂时保存在RabbitMQ中
        //不会由于消费者宕机而造成数据丢失
        //第三个参数：处理数据的回调函数
        //第四个参数：取消接收时的回调函数
        cc.basicConsume("work.queue", false, d, c);
    }
}
